package com.udf.flink.scala.apitest.broadcast

import org.apache.flink.api.common.functions.RichMapFunction
import org.apache.flink.api.scala.ExecutionEnvironment
import org.apache.flink.configuration.Configuration
import scala.collection.mutable.ArrayBuffer

//flink广播变量使用案例
object BroadCast {
	def main(args: Array[String]): Unit = {
		val environment: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
		import org.apache.flink.api.scala._
		//准备数据集
		val userInfo = ArrayBuffer(("zs", 10), ("ls", 20), ("ww", 30))
		//加载数据集构建DataSet--需要广播的数据
		val userDataSet: DataSet[(String, Int)] = environment.fromCollection(userInfo)
		//原始数据
		val data = environment.fromElements("zs", "ls", "ww")
		//在这里需要使用到RichMapFunction获取广播变量
		val result = data.map(new RichMapFunction[String, String] {
			//定义一个list集合，用户接受open方法中获取到的广播变量
			var listData: java.util.List[(String, Int)] = null
			//定义一个map集合，存储广播变量中的内容
			var allMap = Map[String, Int]()

			//初始化方法  可以在open方法中获取广播变量数据
			override def open(parameters: Configuration): Unit = {
				//获取广播变量(broadcastMapName)的值
				listData = getRuntimeContext.getBroadcastVariable[(String, Int)]("broadcastMapName")
				val it = listData.iterator()
				while (it.hasNext) {
					val tuple = it.next()
					allMap += (tuple._1 -> tuple._2)
				}
			}


			//使用广播变量操作数据
			override def map(name: String): String = {
				val age = allMap.getOrElse(name, 20)
				name + "," + age
			}
		}).withBroadcastSet(userDataSet, "broadcastMapName")
		result.print()
	}
}
/*
// 广播状态描述
val broadcastStateDesc: MapStateDescriptor[String, String] =
    new MapStateDescriptor[String, String]("broadcast-desc", classOf[String], classOf[String])
// 将普通的非广播流转为广播流
val ruleStream: BroadcastStream[String] = normalStream.broadcast(broadcastStateDesc)
通过connect算子来将两条流连接在一起，此时广播流ruleStream就会被广播到非广播流actionStream中，得到的是一个BroadcastConnectedStream的流。
BroadcastConnectedStream流本质上包含了广播流ruleStream和非广播流actionStream。
val connectedStream: BroadcastConnectedStream[(String, Int), String] = actionStream.connect(ruleStream)

此时process算子中的参数类型会根据非广播流actionStream的类型分为两种。如果actionStream有经过keyBy算子操作后转为KeyedStream类型
那么process()中为KeyedBroadcastProcessFunction否则为BroadcastProcessFunction。
(此处就只简单使用下KeyedBroadcastProcessFunction，它们两具体的区别和功能大家去参考下官网哈)。
在使用上都有两个方法：processElement处理非connected流数据并且只可读取广播状态，processBroadcastElement处理connectedStream流数据并且可读写广播状态。
因为flink里面没有跨任务通信的机制，在一个任务实例中的修改不能在并行任务间传递。
得保证BroadcastState在算子的并行实例是相同的，所以不能让单个任务去修改状态，只能让广播方修改。

 */